package com.sdgs.paas.sdk.core.stream;

import java.util.Map;
import java.util.Map.Entry;

import javax.annotation.PostConstruct;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.context.ApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.integration.channel.DirectChannel;

/**
 * stream服务
 * 
 * @author wjj
 * 
 */
public class PaasStreamService implements MessageHandler {
	@Autowired
	Source source;
	@Autowired
	Sink sink;
	@Autowired
	ApplicationContext ctx;
	Map<String, IPaasStreamHandler> handlers;

	@PostConstruct
	public void init() {
		DirectChannel dc=((DirectChannel)this.sink.input());
		dc.setComponentName(Sink.INPUT);
		dc.setApplicationContext(ctx);
		dc.subscribe(this);
		handlers = ctx.getBeansOfType(IPaasStreamHandler.class);
	}

	public boolean sendMsg(String msg) {
		return this.source.output().send(MessageBuilder.withPayload(msg).build());
	}

	public Source getSource() {
		return source;
	}

	public void setSource(Source source) {
		this.source = source;
	}

	public Sink getSink() {
		return sink;
	}

	public void setSink(Sink sink) {
		this.sink = sink;
	}

	@Override
	public void handleMessage(Message<?> message) throws MessagingException {
		for (Entry<String, IPaasStreamHandler> entry : handlers.entrySet()) {
			entry.getValue().dealMsg(new String((byte[]) message.getPayload()));
		}

	}

	public ApplicationContext getCtx() {
		return ctx;
	}

	public void setCtx(ApplicationContext ctx) {
		this.ctx = ctx;
	}

	public Map<String, IPaasStreamHandler> getHandlers() {
		return handlers;
	}

	public void setHandlers(Map<String, IPaasStreamHandler> handlers) {
		this.handlers = handlers;
	}

}
